package com.atguigu.gmall.realtime.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.util.DruidDSUtil;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * 单独类
 * 从主流中过滤数据
 *

 */

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    DruidDataSource druidDataSource = null;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //创建连接对象只需要一个并行任务创建一个就好
    //private Connection connection;
    @Override
    public void open(Configuration parameters) throws Exception {
        //获取连接池对象
        druidDataSource = DruidDSUtil.createDataSource();
        //这种方式创建有可能长时间没来数据会关掉，超时时间
        //注册驱动类
        //Class.forName(GmallConfig.PHOENIX_DRIVER);
        //获取连接对象
        //connection = DriverManager.getConnection(GmallConfig.PHOENIX_URL);
    }

    //处理主流数据
    @Override
    public void processElement(JSONObject jsonObject, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //根据业务数据的表名到状态中判断是否存在，如果存在是维度数据，不存在直接过滤
        //jsonobj是Maxwell封装的json字符串封装的jsonobj  包括data 其他属性
        String tableName = jsonObject.getString("table");
        //获取业务数据json对象
        TableProcess tableProcess = broadcastState.get(tableName);
        if (tableProcess != null){
            //说明是维度数据
            JSONObject dataJsonObj = jsonObject.getJSONObject("data");

            //过滤不需要的数据
            String sinkColumns = tableProcess.getSinkColumns();
            filter(dataJsonObj,sinkColumns);

            //将sink——table补充到维度数据里
            String sinkTable = tableProcess.getSinkTable();
            dataJsonObj.put("sink_table",sinkTable);

            out.collect(dataJsonObj);
        }
        else{
            System.out.println("不是维度数据 " + jsonObject);
        }
    }



    //处理广播流数据
    //数据格式json字符串，包括before.after等，after中保存的是数据库中的一条数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
        //将jsonStr转换成jsonObj
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //获取一条数据,转换成tableproce对象
        TableProcess tableProcess = jsonObj.getObject("after", TableProcess.class);

        //获取表中的属性
        String sourceTable = tableProcess.getSourceTable();
        String sinkTable = tableProcess.getSinkTable();
        String sinkPk = tableProcess.getSinkPk();
        String sinkColumns = tableProcess.getSinkColumns();
        String sinkExtend = tableProcess.getSinkExtend();

        //拼接建表语句
        createTable(sinkTable,sinkColumns,sinkPk,sinkExtend);

        //获取创建的状态，不推荐再次创建，通过参数传过来，将配置表信息保存在状态中
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        broadcastState.put(sourceTable,tableProcess);

    }
    //过滤字段属性
    private void filter(JSONObject dataJsonObj , String columns) {
        //将维度表数据中column字段取出来，并转换成list
        String[] columnArr = columns.split(",");
        List<String> columnList = Arrays.asList(columnArr);

        //获取数据中的所有元素
     Set<Map.Entry<String, Object>> entries = dataJsonObj.entrySet();
     entries.removeIf(entry -> !columnList.contains(entry.getKey()));


/*
        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while (iterator.hasNext()){
            Map.Entry<String, Object> next = iterator.next();
            if (!columnList.contains(next.getKey())){
                iterator.remove();
            }

        }
*/



    }

    //创建维度表方法
    private void createTable(String sinkTable, String columnStr, String sinkPk, String ext) {
        //对ext空值处理
        if (ext ==null){
            ext = "";
        }

        if (sinkPk == null){
            sinkPk = "id";
        }
        //拼接建表语句,注意空格
        StringBuilder createSql = new StringBuilder("create table if not exists "+ GmallConfig.PHOENIX_SCHEMA +"."+sinkTable+"(");
        String[] columnArr = columnStr.split(",");
        for (int i = 0; i < columnArr.length; i++) {
            String column = columnArr[i];
            if (column.equals(sinkPk)){
                createSql.append(column + " varchar primary key");
            }else{
                createSql.append(column + " varchar");
            }

            if (i < columnArr.length - 1){
                createSql.append(",");
            }

        }
        createSql.append(") " + ext);
        System.out.println("建表语句是：" + createSql);

        // TODO: 2022/5/18  在phoenix中执行sql
        Connection connection = null;
        PreparedStatement ps = null;
        try {
            //从连接池获取连接对象
            connection = druidDataSource.getConnection();
            //创建操作对象
             ps = connection.prepareStatement(createSql.toString());
            //执行sql语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }finally {
            //释放资源
            if (ps != null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }

            if (connection != null){
                try {
                    connection.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
